package net.wlm.others;

import com.alibaba.fastjson.JSON;
import net.wlm.utils.JdbcUtil;
import net.wlm.utils.KafkaUtil;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;

public class VehicleAbnormalAlarmJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "vehicle-date-geohash");
        DataStream<VehicleData> vehicleStream = kafkaStream.map(o -> JSON.parseObject(o, VehicleData.class));

        // 定义电池温度异常模式（连续3次超过60度）
        Pattern<VehicleData, ?> batteryTempPattern = Pattern.<VehicleData>begin("first")
                .where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) {
                        return value.getBatteryTemp() > 60.0;
                    }
                })
                .next("second").where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) {
                        return value.getBatteryTemp() > 60.0;
                    }
                })
                .next("third").where(new SimpleCondition<VehicleData>() {
                    @Override
                    public boolean filter(VehicleData value) {
                        return value.getBatteryTemp() > 60.0;
                    }
                })
                .within(Time.minutes(5)); // 5分钟内连续3次

        // 应用CEP模式
        PatternStream<VehicleData> patternStream = CEP.pattern(
                vehicleStream.keyBy(VehicleData::getVin),
                batteryTempPattern
        );

        // 处理匹配事件
        DataStream<String> alarmStream = patternStream.process(new PatternProcessFunction<VehicleData, String>() {
            @Override
            public void processMatch(
                    Map<String, List<VehicleData>> matches,
                    Context ctx,
                    Collector<String> out) throws Exception {
                VehicleData first = matches.get("first").get(0);
                VehicleData third = matches.get("third").get(0);

                String alarmMsg = String.format("车辆%s连续3次电池温度超标(%.1f°C->%.1f°C)",
                        first.getVin(),
                        first.getBatteryTemp(),
                        third.getBatteryTemp());
                out.collect(alarmMsg);
            }
        });

        alarmStream.print("温度报警:");
        JdbcUtil.sinkToClickhouseUpsert(alarmStream,
                "INSERT INTO vehicle_log.vehicle_abnormal_alarm VALUES (?, ?, ?, ?)");

        env.execute("VehicleAbnormalAlarmCEPJob");
    }
}